//
// Created by LiuYou on 2021/8/15.
//

#include <cstdlib>
#include <fcntl.h>
#include <sys/epoll.h>
#include <cstdio>
#include <cstring>
#include <sys/socket.h>
#include <unistd.h>
#include <cerrno>
#include <netinet/in.h>
#include <arpa/inet.h>
#include <thread>


class WorkerThreadArgs {
public:
    int epollFd;
    // 这里的 sockFd其实就是 connectFd。
    int sockFd;

};

static int setNonblocking(int fd) {
    int oldOption = fcntl( fd, F_GETFL );
    int newOption = oldOption | O_NONBLOCK;
    fcntl( fd, F_SETFL, newOption );
    return oldOption;
}


static void addEvent(int epollFd, int fd, bool oneShot) {
    epoll_event epollEvent{};
    epollEvent.data.fd = fd;
    epollEvent.events = EPOLLIN | EPOLLET;
    if (oneShot) {
        epollEvent.events |= EPOLLONESHOT;
    }
    epoll_ctl( epollFd, EPOLL_CTL_ADD, fd, &epollEvent );
    setNonblocking( fd );
}

static void resetOneShot(int epollFd, int fd) {
    epoll_event epollEvent{};
    epollEvent.data.fd = fd;
    epollEvent.events = EPOLLIN | EPOLLET | EPOLLONESHOT;
    epoll_ctl( epollFd, EPOLL_CTL_MOD, fd, &epollEvent );
}

static void* workerThread(void* arg) {
    int sockFd = ((WorkerThreadArgs*) arg)->sockFd;
    int epollFd = ((WorkerThreadArgs*) arg)->epollFd;
//    printf( "start new thread to receive data on fd: %d\n", connectFd );
    printf( "创建一个新的线程, 线程ID: %lu, 从fd: %d 接受数据\n",
            pthread_self(),
            sockFd );
    char buffer[1024];
    memset( buffer, '\0', sizeof( buffer ));
    // 循环读取sockFd上的数据, 知道遇到EAGAIN错误。
    while (true) {
        ssize_t ret = recv( sockFd, buffer, 1024 - 1, 0 );
        if (ret == 0) {
            close( sockFd );
            printf( "客户端关闭了连接\n" );
            break;
        } else if (ret < 0) {
            // 文件描述符是非阻塞的, 在处理完后, 没有数据来, 这个文件描述符不是就绪态。
            // 如果确实不是就绪态, 就执行下面的if语句。重置该 connectFd上的注册的事件。
            if (errno == EAGAIN) {
                resetOneShot( epollFd, sockFd );
                printf( "读完了, 不再是就绪态。\n" );
                break;
            }
        } else {
//            printf( "get content: %s\n", buffer );
            printf( "得到内容为: %s\n", buffer );
            fflush( stdout );
            // 休眠5秒, 模拟数据处理过程。
            sleep( 5 );
        }
    }
//    printf( "end thread receiving data on fd: %d\n", connectFd );
    printf( "线程结束。" );

    pthread_exit( nullptr );
}


/**
 * @brief 使用 EPOLLONESHOT事件。
 * @param argc 
 * @param argv 
 * @return 
 */
int main(int argc, char** argv) {
    if (argc <= 2) {
        printf( "用法: %s ip port\n", basename( argv[0] ));
        exit( 1 );
    }

    char* ip = argv[1];
    int port = atoi( argv[2] );

    int listenFd = socket( PF_INET, SOCK_STREAM, 0 );
    // 判断

    // 处理异常终止, 端口没有立刻释放的问题。这样就可以下一次继续bind()这个端口。
    int optVal = 1;
    if (setsockopt( listenFd, SOL_SOCKET, SO_REUSEADDR, &optVal, sizeof( optVal )) == -1) {
        perror( "setsockopt()" );
        exit( 1 );
    }

    sockaddr_in localAddress{};
    localAddress.sin_family = AF_INET;
    inet_pton( AF_INET, ip, &localAddress.sin_addr );
    localAddress.sin_port = htons( port );
    int ret = bind( listenFd, (sockaddr*) &localAddress, sizeof( localAddress ));
    if (ret == -1) {
        perror( "bind()" );
        close( listenFd );
        exit( 1 );
    }

    ret = listen( listenFd, 5 );
    if (ret == -1) {
        perror( "listen()" );
        close( listenFd );
        exit( 1 );
    }

    epoll_event epollEvents[1024]{};
    int epollFd = epoll_create( 5 );
    // 判断

    // 这里并没有给listenFd注册 EPOLLONESHOT事件。
    addEvent( epollFd, listenFd, false );

    while (true) {
        int readyFd = epoll_wait( epollFd, epollEvents, 1024, -1 );
        if (readyFd < 0) {
            printf( "epoll failure\n" );
            break;
        }

        for (int i = 0; i < readyFd; ++i) {
            if (epollEvents[i].data.fd == listenFd) {
                sockaddr_in remoteAddress{};
                socklen_t socklen = sizeof( remoteAddress );
                int connectFd = accept( epollEvents[i].data.fd, (sockaddr*) &remoteAddress, &socklen );
                if (connectFd == -1) {
                    perror( "accept()" );
                    break;
                }
                // 对每个非监听文件描述符都注册 EPOLLONESHOT事件。
                addEvent( epollFd, connectFd, true );
            } else if (epollEvents[i].events & EPOLLIN) {
                WorkerThreadArgs fdsForNewWorker{};
                fdsForNewWorker.epollFd = epollFd;
                fdsForNewWorker.sockFd = epollEvents[i].data.fd;
                // 新启动一个工作线程为 connectFd服务。
                pthread_t tid;
                // 这里注意: 当这个线程读取完数据后, 开始处理这些数据时, connectFd又有新数据了, 这时触发了 EPOLLIN事件,
                // 但是因为我们为connectFd注册 EPOLLONESHOT事件, 所以, 在一个线程在处理某个socket时, 其它线程是不可能有机会
                // 操作这个socket的。
                //
                pthread_create( &tid, nullptr, workerThread, (void*) &fdsForNewWorker );
//                sleep( 10 );
                // 这里要注意: pthread_join()调用要放在哪?
                pthread_join( tid, nullptr );
            } else {
                printf( "something else happened\n" );
            }
        }
    }

    close( listenFd );

    exit( 0 );
}